package com.tuanzi.loan.business.service.family.impl;

import com.google.common.collect.Lists;
import com.tuanzi.loan.business.entity.common.CommProductId;
import com.tuanzi.loan.business.service.common.CommProductIdService;
import com.tuanzi.loan.business.service.family.SwitchService;
import com.tuanzi.loan.core.aspect.Debug;
import com.tuanzi.loan.core.vo.PageSearcher;
import com.tuanzi.loan.vo.request.family.SwitchListSearchRequest;
import com.tuanzi.loan.vo.request.family.SwitchRank;
import com.tuanzi.loan.vo.response.family.SwitchListSearchResponse;
import com.tuanzi.loan.vo.view.family.SwitchInfoView;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.Pageable;
import org.springframework.data.redis.core.BoundSetOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.retry.RetryException;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Debug
@Service
@Transactional
@Retryable(value = { RetryException.class })
public class SwitchServiceImpl implements SwitchService {

	@Autowired
	private StringRedisTemplate stringRedisTemplate;

	@Autowired
	private CommProductIdService commProductIdService;

	// 目前大产品只有菠萝贷和快刻在使用这些开关，值来自字典 LARGE_COMMON_PRODUCT：1 代表菠萝贷、9 代表熊花花、10 代表快刻周转
	private static final String[] DEFAULT_PARENT_PRODUCTS = new String[] { "1", "9", "10" };

	private static final Comparator<SwitchListSearchResponse> SWITCH_RESPONSE_COMPARATOR = (before, after) -> {
		int[] results = new int[] {
			before.getParentProductId().compareTo(after.getParentProductId()),
			before.getSwitchModule().compareTo(after.getSwitchModule()),
			before.getSwitchName().compareTo(after.getSwitchName())
		};

		for (int result : results) {
			if (result != 0) {
				return result;
			}
		}

		// 不太可能走到这里
		return 0;
	};

	@Override
	public Page<SwitchListSearchResponse> list(PageSearcher<SwitchListSearchRequest> pageSearcher, HashMap<String, SwitchRank> switchRankMap) {
		// 1.限定查询范围
		SwitchListSearchRequest request = pageSearcher.getSearchCondition();

		String[] parentProducts = null;
		// 用户指定了大产品
		if (StringUtils.isNotEmpty(request.getParentProductId()) && ArrayUtils.contains(DEFAULT_PARENT_PRODUCTS, request.getParentProductId())) {
			parentProducts = new String[] { request.getParentProductId() };
		} else {
			parentProducts = DEFAULT_PARENT_PRODUCTS;
		}

		List<SwitchRank> rankList = null;
		// 用户指定了开关模块
		if (StringUtils.isNotEmpty(request.getSwitchModule())) {
			rankList = Lists.newArrayList(switchRankMap.get(request.getSwitchModule()));
		} else {
			rankList = Lists.newArrayList(switchRankMap.values());
		}

		// 2.从限定范围中返回对应开关记录
		// Map: productId - CommProductId
		Map<Integer, CommProductId> commProductIdMap = commProductIdService.findAll().stream().collect(Collectors.toMap(CommProductId::getProductId, commProductId -> commProductId));
		// Map: parentId - (switchMode#switchName - Response)  ==》 之所以使用switchMode#switchName为key是为了防止开关名重名，故使用开关模块和开关名作为key
		HashMap<String, HashMap<String, SwitchListSearchResponse>> parentProduct2Switch = new HashMap<>();
		for (String parentProduct : parentProducts) {
			parentProduct2Switch.put(parentProduct, new HashMap<>());
		}

		// 遍历开关分组（模块）
		for (SwitchRank rank : rankList) {
			String switchMode = rank.rankId; // 开关模块
			// 遍历开关分组下的成员（开关名）
			for (String switchName : rank.members) {
				// 如果指定了开关名但当前开关名与指定开关名不一致，跳过
				if (StringUtils.isNotEmpty(request.getSwitchName()) && !switchName.equals(request.getSwitchName())) {
					continue;
				}
				// 换言之，在没指定开关名或指定了开关名且当前开关名与指定开关名一致的情况下都满足查询条件

				// 当前开关名满足查询条件，为其预置大产品范围内的 Response
				for (String parentProduct : parentProducts) {
					parentProduct2Switch.get(parentProduct).put(String.format("%s#%s", switchMode, switchName), new SwitchListSearchResponse(parentProduct, switchMode, switchName));
				}

				if (!stringRedisTemplate.hasKey(switchName)) {
					continue;
				}
				BoundSetOperations<String, String> setOps = stringRedisTemplate.boundSetOps(switchName);
				// 得到对应开关下开启的产品 id 集合,若没有任何产品开启过，跳过
				Set<String> switchOns = setOps.members();
				if (CollectionUtils.isEmpty(switchOns)) {
					continue;
				}

				List<Integer> productIds = switchOns.stream().map(Integer::valueOf).collect(Collectors.toList());
				// 遍历产品 id 集合，将产品 id 放到对应的 Response 中
				for (Integer productId : productIds) {
					String parentId = commProductIdMap.get(productId).getParentId();
					if (parentProduct2Switch.containsKey(parentId)) {
						SwitchListSearchResponse response = parentProduct2Switch.get(parentId).get(String.format("%s#%s", switchMode, switchName));
						response.addProductId(productId);
					}
				}
			}
		}

		List<SwitchListSearchResponse> list = Lists.newArrayList();
		for (HashMap<String, SwitchListSearchResponse> map : parentProduct2Switch.values()) {
			list.addAll(map.values());
		}
		// 将 Response 排序便于分页
		list.sort(SWITCH_RESPONSE_COMPARATOR);

		Pageable pageable = pageSearcher.getPageable();
		int toIndex = Math.min(pageable.getOffset() + pageable.getPageSize(), list.size());

		return new PageImpl<>(list.subList(pageable.getOffset(), toIndex), pageable, list.size());
	}

	@Override
	public List<String> findBy(SwitchInfoView switchInfo) {
		if (!stringRedisTemplate.hasKey(switchInfo.getSwitchName())) {
			return Lists.newArrayList();
		}

		BoundSetOperations<String, String> setOps = stringRedisTemplate.boundSetOps(switchInfo.getSwitchName());
		Set<String> productIds = setOps.members();
		if (CollectionUtils.isEmpty(productIds)) {
			return Lists.newArrayList();
		}

		Map<Integer, CommProductId> commProductIdMap = commProductIdService.findAll().stream().collect(Collectors.toMap(CommProductId::getProductId, commProductId -> commProductId));
		List<String> list = Lists.newArrayList();
		for (String productId : productIds) {
			String parentId = commProductIdMap.get(Integer.valueOf(productId)).getParentId();
			if (parentId.equals(switchInfo.getParentProductId())) {
				list.add(productId);
			}
		}
		return list;
	}

	@Override
	public void openSwitch(SwitchInfoView switchInfo) {
		List<String> on = switchInfo.getProductIds();
		List<String> off = commProductIdService.findAll().stream().filter(commProductId -> commProductId.getParentId().equals(switchInfo.getParentProductId()) && !on.contains(commProductId.getProductId().toString())).map(commProductId -> commProductId.getProductId().toString()).collect(Collectors.toList());
		BoundSetOperations<String, String> setOps = stringRedisTemplate.boundSetOps(switchInfo.getSwitchName());
		if (CollectionUtils.isNotEmpty(on)) {
			setOps.add(on.toArray(new String[0]));
		}
		if (CollectionUtils.isNotEmpty(off)) {
			setOps.remove(off.toArray());
		}
	}
}
